package com.aispeech.jarpc;

import com.aispeech.jarpc.message.ArpcMessage;
import com.aispeech.jarpc.message.EMsgType;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.NoConnectionPendingException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

@Slf4j
public class TcpClient {
    private JArpc rpc;
    private SocketChannel socketChannel = null;

    private boolean isAsyncMode;

    private int sn;
    private SendState sendState;
    private RecvState recvState;
    private ArpcMessage messageRecving;
    private ArpcException exception;
    private ByteBuffer headerBuffer;
    private ByteBuffer bodyBuffer;
    private Object lock;

    public TcpClient(JArpc rpc) {
        this.rpc = rpc;
        lock = new Object();

        isAsyncMode = rpc.getCallback() != null;
        sendState = SendState.START;
    }

    public ArpcException getException() {
        return exception;
    }

    public ArpcMessage getMessageRecving() {
        return messageRecving;
    }

    public void wait(int timeout) {
        synchronized (lock) {
            try {
                lock.wait(timeout);
            } catch (InterruptedException e) {
                log.error(String.format("session: %s, timeout value: %d", new String(rpc.getSessionId(), Charset.defaultCharset()), timeout));
                exception = new ArpcException(String.format("[ %s ]: %s", e.getClass().getName(), e.getMessage()));
            }
        }
    }

    public void connect(String host, int port) {
        try {
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
//        log.info("connect to: {} ...", host);
            String ip = InetAddress.getAllByName(host)[0].getHostAddress();
//        log.info("2. connect to ip: {} ...", ip);
            JArpc.Selector().wakeup();
            socketChannel.register(JArpc.Selector(), SelectionKey.OP_CONNECT, this);
            socketChannel.connect(new InetSocketAddress(ip, port));
//            log.info("begin connect to host: {}, {}:{}...", host, ip, port);
        } catch (IOException e) {
            exception = new ArpcException(e.getMessage());

            synchronized (lock) {
                lock.notifyAll();
            }
        }
    }

    public void close() throws IOException {
        if (isConnected()) {
            socketChannel.close();
        }
        socketChannel = null;
    }

    public boolean isConnected() {
        return (socketChannel != null && socketChannel.isConnected());
    }

    public void send(ByteBuffer data) throws IOException {
        while (data.hasRemaining()) {
            socketChannel.write(data);
        }
//        log.info("[ REQ ] msg(sn:{}, length: {}) sent.", sn, data.capacity());
        sn += 1;

        //messageRecving = null;
    }

    public void sendWithSessionId(String params, byte[] sessionId) throws IOException {
        ByteArrayOutputStream stream = new ByteArrayOutputStream(params.length() + sessionId.length);
        stream.write(params.getBytes());
        stream.write(sessionId);

        send(new ArpcMessage(EMsgType.JSON, sn, stream.toByteArray()).asByteBuffer());
    }

    public void sendJson(String json) throws IOException {
        send(new ArpcMessage(EMsgType.JSON, sn, json.getBytes()).asByteBuffer());
    }

    public void sendBinary(byte[] data, int len) throws IOException {
        EMsgType msgType = len > 0 ? EMsgType.BINARY : EMsgType.STOP;
        sendState = len > 0 ? SendState.FEED : SendState.STOP;
        send(new ArpcMessage(msgType, sn, data, len).asByteBuffer());
    }

    public void sendProcessRequest(byte[] sessionId, ByteBuffer data) throws IOException {
        byte[] req = data.array();
        ByteArrayOutputStream stream = new ByteArrayOutputStream(req.length + sessionId.length);
        stream.write(req);
        stream.write(sessionId);

        send(new ArpcMessage(EMsgType.PROCESS, sn, stream.toByteArray()).asByteBuffer());
    }

    public void onConnected(SelectionKey selectionKey, SocketChannel socketChannel) throws IOException {
        socketChannel.configureBlocking(false);

        sn = 0;
        recvState = RecvState.HEADER;
        headerBuffer = ByteBuffer.allocate(ArpcMessage.MESSAGE_HEADER_LENGTH);
        messageRecving = null;
        exception = null;

        try {
            socketChannel.finishConnect();
            socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ, this);
        } catch (ConnectException | NoConnectionPendingException ex) {
            exception = new ArpcException(EArpcState.S_CONNECT_FAILED.getValue(), String.format("Failed connect to %s:%d", rpc.getHost(), rpc.getPort()));
        }

//        log.info("onConnected.");

        synchronized (lock) {
            lock.notifyAll();
        }
    }

    public void onReceived(SelectionKey selectionKey, SocketChannel socketChannel) throws IOException {
        ByteBuffer byteBuffer = recvState == RecvState.HEADER ? headerBuffer : bodyBuffer;

        long bytesRead = socketChannel.read(byteBuffer);
        if (bytesRead == -1) {
            socketChannel.close();

            exception = new ArpcException(EArpcState.S_CONNECTION_BROKEN);

            if (isAsyncMode && SendState.FEED == sendState)
                rpc.getCallback().onArpcError(exception);
            else {
                synchronized (lock) {
                    lock.notifyAll();
                }
            }
        } else if (bytesRead > 0) {
            if (byteBuffer.position() >= byteBuffer.capacity()) {
                if (recvState == RecvState.HEADER) {
                    parseMessageHeader();
                } else {
                    parseMessageBody();
                }

                byteBuffer.clear();
            }

            selectionKey.interestOps(SelectionKey.OP_READ);
        }
    }

    private void parseMessageBody() {
    	if(messageRecving == null ) {
    		log.info("[ parseMessageBody ] messageRecving :{}");
    	}
    	if(bodyBuffer == null) {
    		log.info("[ parseMessageBody ] bodyBuffer is null");
    	}
    	
        messageRecving.setBody(bodyBuffer.array());

        if (messageRecving.getMsgType() == EMsgType.ERROR) {
            exception = new ArpcException(new String(messageRecving.getBody()));

            if (isAsyncMode && SendState.FEED == sendState) { // start消息的走同步方法
                rpc.getCallback().onArpcError(exception);
            }
        } else {
            if (isAsyncMode && SendState.FEED == sendState) {
                rpc.getCallback().onArpcMessage(messageRecving);
            }

            exception = null;
        }

        synchronized (lock) {
            lock.notifyAll();
        }

        recvState = RecvState.HEADER;
    }

    private void parseMessageHeader() {
//        log.info("[ parseMessageHeader ] recv length: {}", headerBuffer.position());

        messageRecving = ArpcMessage.from(headerBuffer.array());

        if (messageRecving.getLength() > ArpcMessage.MAX_MESSAGE_SIZE) {
            exception = new ArpcException(EArpcState.S_MSG_OVERLOAD);

            if (isAsyncMode && SendState.FEED == sendState) {
                rpc.getCallback().onArpcError(exception);
            } else {
                synchronized (lock) {
                    lock.notifyAll();
                }
            }
        } else {
            recvState = RecvState.BODY;
            bodyBuffer = ByteBuffer.allocate(messageRecving.getLength());
        }
    }

    enum RecvState {HEADER, BODY}
    enum SendState {START, FEED, STOP}
}
